package org.example.table_type;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

public class FlinkBatchObjectTable {
    public static void main(String[] args) throws Exception {
        // 使用批处理模式创建TableEnvironment
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .inBatchMode() // 设置为批处理模式
                .build();
        TableEnvironment tableEnv = TableEnvironment.create(settings);

        // 创建Paimon catalog，使用默认文件系统
        String createCatalogSQL = "CREATE CATALOG paimon WITH (\n" +
                "    'type' = 'paimon',\n" +
                "    'warehouse' = 'file:///tmp/paimon'\n" +
                ");";
        System.out.println("执行SQL: \n" + createCatalogSQL);
        tableEnv.executeSql(createCatalogSQL);

        // 先删除表（如果存在）
        String dropTableSQL = "DROP TABLE IF EXISTS `paimon`.`default`.`t_user_files`;";
        System.out.println("执行SQL: \n" + dropTableSQL);
        tableEnv.executeSql(dropTableSQL);

        // 创建Paimon对象表
        String createTableSQL = "CREATE TABLE `paimon`.`default`.`t_user_files` WITH (\n" +
                "  'type' = 'object-table',\n" +
                "  'object-location' = 'file:///tmp/paimon/default.db/t_user' \n" +
                ");";
        System.out.println("执行SQL: \n" + createTableSQL);
        tableEnv.executeSql(createTableSQL);

        tableEnv.executeSql("use catalog `paimon`");

        String refreshTable = "CALL sys.refresh_object_table('default.t_user_files');";
        System.out.println("执行SQL: \n" + refreshTable);
        System.out.println("刷新表:");
        tableEnv.executeSql(refreshTable).print();

        // 查询表信息
        String descTableSQL = "SELECT * FROM `paimon`.`default`.`t_user_files`;";
        System.out.println("执行SQL: \n" + descTableSQL);
        tableEnv.executeSql(descTableSQL).print();
        
        // 确保所有操作都已经完成
        System.out.println("查询执行完毕");
    }
}